package com.bw.app.dws;

import com.bw.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Dws_The_populariy_is_new {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        env.setParallelism(1);

        String topic = "dwd_page_action_order_detail";
        String groupId = "dws_shop_popularity_analysis";

        // 1. 创建Kafka源表，连接到页面行为和订单详情主题
        tableEnv.executeSql("CREATE TABLE page_action_source (" +
                "  shop_id STRING COMMENT '店铺ID', " +
                "  mid STRING COMMENT '设备唯一标识', " +
                "  user_id STRING COMMENT '用户ID', " +
                "  is_new STRING COMMENT '用户ID', " +
                "  ts_new TIMESTAMP(3) COMMENT '页面操作时间戳', " +
                "  order_id STRING COMMENT '订单ID', " +
                "  consignee STRING COMMENT '收货人姓名', " +
                "  consignee_tel STRING COMMENT '收货人电话', " +
                "  total_amount STRING COMMENT '订单总金额', " +
                "  order_status STRING COMMENT '订单状态码', " +
                "  payment_way STRING COMMENT '支付方式', " +
                "  delivery_address STRING COMMENT '配送地址', " +
                "  out_trade_no STRING COMMENT '外部交易号', " +
                "  order_create_time STRING COMMENT '订单创建时间', " +
                "  order_detail_id STRING COMMENT '订单详情ID（主键）', " +
                "  sku_id STRING COMMENT '商品SKU ID', " +
                "  sku_name STRING COMMENT '商品名称', " +
                "  order_price STRING COMMENT '商品单价', " +
                "  sku_num STRING COMMENT '商品数量', " +
                "  split_total_amount STRING COMMENT '分摊后金额', " +
                "  split_activity_amount STRING COMMENT '分摊活动优惠金额', " +
                "  split_coupon_amount STRING COMMENT '分摊优惠券金额', " +
                "  cart_id STRING COMMENT '购物车ID', " +
                "  is_ordered_num STRING COMMENT '是否加购', " +
                "  cart_sku_num STRING COMMENT '购物车中商品数量', " +
                "  cart_create_time STRING COMMENT '加入购物车时间', " +
                "  favor_id STRING COMMENT '收藏ID', " +
                "  is_favor_canceled STRING COMMENT '是否取消收藏', " +
                "  favor_create_time STRING COMMENT '收藏创建时间', " +
                "  row_op_ts TIMESTAMP_LTZ(3) COMMENT '数据处理时间戳', " +
                "  WATERMARK FOR row_op_ts AS row_op_ts - INTERVAL '5' SECONDS " +
                ")" + MyKafkaUtil.getKafkaDDL(topic,groupId));

        // 2. 定义店铺人气指标计算逻辑
        tableEnv.executeSql("SELECT " +
                        "  shop_id, " +
                        "  sku_id, " +
                        "  is_new, " +
                        "  window_start, " +
                        "  window_end, " +
                        "  COUNT(CASE WHEN is_new IS NOT NULL THEN 1 ELSE 0 END) AS is_new, " +//-- 新老用户
                        "  COUNT(DISTINCT mid) AS visitor_count, " +//-- 独立访问设备数
                        "  COUNT(ts_new) AS search_count, " +//新搜索次数
                        "  SUM(CASE WHEN is_ordered_num IS NOT NULL THEN 1 ELSE 0 END) AS add_to_cart_count, " +//加购次数
                        "  COUNT(favor_id) AS favorite_count, " +//有效收藏次数
                        "  SUM(CASE WHEN order_status IN ('1002', '1004') THEN 1 ELSE 0 END) AS payment_count, " +//支付次数（订单状态为已支付或已完成）
                        "  COUNT(out_trade_no) AS member_join_count, " +//入会次数
                        "  COUNT(DISTINCT ts_new) AS consult_count " +//咨询次数
                        "FROM TABLE(" +
                        "  TUMBLE(TABLE page_action_source, DESCRIPTOR(row_op_ts), INTERVAL '10' SECONDS)" + // 定义10秒滚动窗口
                        ") " +
                        "GROUP BY shop_id , sku_id, is_new ,window_start, window_end")
                .print();

        env.execute();
    }
}
